package net.bwie.realtime.jtp.ods.Function;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.configuration.Configuration;

public class CdcDataEtlMapFunction extends RichMapFunction<String, String> {


    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
    }

    /**
     *
     * 1 - 解析json
     * 2 - 获取字段值
     * 3 - 拼接json字符串
     * 4 - 返回
     */
    @Override
    public String map(String value) throws Exception {
        //1 - 解析json
        JSONObject jsonObject = JSON.parseObject(value);

        //2 - 获取字段after
        JSONObject afterValue = jsonObject.getJSONObject("after");
        //获取字段before
        JSONObject beforeValue = jsonObject.getJSONObject("before");
        //获取字段 op
        String opValue = jsonObject.getString("op");
        //获取字段 ts_ms
        Long tsValue = jsonObject.getLong("ts_ms");
        //获取字段 table,db
        JSONObject source = jsonObject.getJSONObject("source");
        String tableValue = source.getString("table");
        String dbValue = source.getString("db");


        String operatorValue;
        switch (opValue){
            case "u":
                operatorValue = "update";
                break;
            case "c":
            case "r":
                operatorValue = "insert";
                break;
            case "d":
                operatorValue = "delete";
                break;
            default:
                operatorValue = "unknown";
        }

        //3 - 拼接json字符串
        JSONObject newObject = new JSONObject();
        newObject.put("db_name",dbValue);
        newObject.put("table_name",tableValue);
        newObject.put("operate_type",operatorValue);
        newObject.put("ts_ms",tsValue);

        if ("d".equals(opValue)){
            newObject.put("operate_data",beforeValue);
        }else{
            newObject.put("operate_data",afterValue);
        }

        return newObject.toJSONString();
    }

    @Override
    public void close() throws Exception {
        super.close();
    }
}
